package com.atguigu.dga.governance.assessor.calc;


import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.dga.common.utils.HttpUtil;
import com.atguigu.dga.ds.bean.TDsTaskInstance;
import com.atguigu.dga.governance.assessor.Assessor;
import com.atguigu.dga.governance.bean.AssessParam;
import com.atguigu.dga.governance.bean.GovernanceAssessDetail;
import com.atguigu.dga.governance.bean.GovernanceMetric;
import org.apache.commons.lang3.time.DateFormatUtils;
import org.apache.commons.lang3.time.DateUtils;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import java.math.BigDecimal;
import java.math.RoundingMode;
import java.text.ParseException;
import java.util.*;
import java.util.stream.Collectors;

@Component("IS_DATA_SKEW")
public class IsDataSkewAssessor extends Assessor {

    @Value("${spark-history.server.url}")
    String sparkHistoryServerUrl;

    private String apiPath="/api/v1/applications/";



    //检查是否有数据倾斜，如果某个stage的最大任务耗时超过其他任务平均耗时耗时的{percent}%, 只检查耗时超过{stage_dur_seconds}秒的stage。
    //存在倾斜给0分，不存在给10分。
    //1    获得当天的任务实例 提取其中的 app_link
    //2    拼接路径 访问http://hadoop102:18080/api/v1/applications/application_1684083580862_0012    获取attempId 要成功的
    //3 访问http://hadoop102:18080/api/v1/applications/application_1684083580862_0012/1/stages
    //      获得stageIds   List<String>   过滤掉小于{stage_dur_seconds} stage
    //4  遍历该list
    //        http://hadoop102:18080/api/v1/applications/application_1684083580862_0012/1/stages/1
    // 获得task任务信息集合 每个任务的duration
    //      求最大任务的耗时， 其他任务的平均耗时     二者   最大任务耗时-其他平均任务耗时/ 其他平均耗时  = real percent  和参考percent 进行对比    超过的给0分
    //
    @Override
    public void checkProblem(AssessParam assessParam, GovernanceAssessDetail governanceAssessDetail) throws ParseException {
        //参数

        JSONObject paramJsonObj = JSON.parseObject(assessParam.getGovernanceMetric().getMetricParamsJson());
        BigDecimal paramPercent = paramJsonObj.getBigDecimal("percent");
        Long stageDurSeconds = paramJsonObj.getLong("stage_dur_seconds");

        //1    获得当天的任务实例 提取其中的 app_link
        List<TDsTaskInstance> tDsTaskInstanceList = assessParam.getTDsTaskInstanceList();
        if(tDsTaskInstanceList ==null){
            return;
        }
        String applink=null;
        //找到当天的任务实例 applink
        for (TDsTaskInstance tDsTaskInstance : tDsTaskInstanceList) {
            if(assessParam.getTableMetaInfo().getTableName().equals("dws_trade_user_order_1d")){
                System.out.println(tDsTaskInstance.getAppLink());
            }
            String taskDt = DateFormatUtils.format(tDsTaskInstance.getSubmitTime(), "yyyy-MM-dd");
            if(taskDt.equals(assessParam.getAssessDate())&& tDsTaskInstance.getState()==7){
                applink=tDsTaskInstance.getAppLink();
            }
        }
        //2    拼接路径 访问http://hadoop102:18080/api/v1/applications/application_1684083580862_0012    获取attempId 要成功的
        String jobUrl=sparkHistoryServerUrl+apiPath+applink;
        String jobJson = HttpUtil.get(jobUrl);
        JSONObject jobJsonObj = JSON.parseObject(jobJson);
        JSONArray attemptsJsonArray = jobJsonObj.getJSONArray("attempts");
        String attempId=null;
        for (int i = 0; i < attemptsJsonArray.size(); i++) {
            JSONObject attemptJsonObj = attemptsJsonArray.getJSONObject(i);
            if(attemptJsonObj.getBoolean("completed")){
                attempId=attemptJsonObj.getString("attemptId");
            }
        }
        //3 访问http://hadoop102:18080/api/v1/applications/application_1684083580862_0012/1/stages
        //      获得stageIds   List<String>   过滤掉小于{stage_dur_seconds} stage
        String stagesUrl=sparkHistoryServerUrl+apiPath+applink+"/"+attempId+"/stages";

        String stagesJson = HttpUtil.get(stagesUrl);
        List<JSONObject> stagesJsonObjList = JSON.parseArray(stagesJson, JSONObject.class);
        // 遍历该list  1 过滤 ： 保留耗时超过{stage_dur_seconds}秒的stage 去除numTasks为1     2 转换提取 stageId
        List<String> stageIdsList = stagesJsonObjList.stream().filter(stageJsonObj -> stageJsonObj.getInteger("numTasks") > 1)
                .filter(stageJsonObj -> {
                    String submissionTimeStr = stageJsonObj.getString("submissionTime");
                    String completionTimeStr = stageJsonObj.getString("completionTime");
                    //2023-05-01T19:27:58.087GMT
                    Date submissionTime = null;
                    Date completionTime =null;
                    try {
                        submissionTime = DateUtils.parseDate( submissionTimeStr,"yyyy-MM-dd'T'HH:mm:ss.SSS'GMT'");
                        completionTime = DateUtils.parseDate( completionTimeStr,"yyyy-MM-dd'T'HH:mm:ss.SSS'GMT'");
                    } catch (ParseException e) {
                        throw new RuntimeException(e);
                    }


                    Long durationMs = completionTime.getTime() - submissionTime.getTime();
                    return durationMs / 1000 > stageDurSeconds;
                }).map(stageJsonObj -> stageJsonObj.getString("stageId")).collect(Collectors.toList());
        //4  遍历该list
        //      访问  http://hadoop102:18080/api/v1/applications/application_1684083580862_0012/1/stages/1
        List<JSONObject> skewStagesList=new ArrayList<>();
        for (String stageId : stageIdsList) {
            String stageSingleUrl=sparkHistoryServerUrl+apiPath+applink+"/"+attempId+"/stages/"+stageId;
            String stageSingleArrJson = HttpUtil.get(stageSingleUrl);
            //只有一个元素的List
            List<JSONObject> stageSingeArrJsonList = JSON.parseArray(stageSingleArrJson, JSONObject.class);
            JSONObject stageSingleJsonObj =  stageSingeArrJsonList.get(0) ;
            JSONObject taskListJsonObj = stageSingleJsonObj.getJSONObject("tasks");
            // 获得task任务信息集合 每个任务的duration
            List<Integer> taskDurationList  =new ArrayList<>();
            for (Object taskObj : taskListJsonObj.values()) {
                JSONObject taskJsonObj = (JSONObject) taskObj;
                Integer duration = taskJsonObj.getInteger("duration");
                taskDurationList.add(duration);
            }
            //      求最大任务的耗时， 其他任务的平均耗时     二者   最大任务耗时-其他平均任务耗时/ 其他平均耗时  = real percent  和参考percent 进行对比    超过的给0分
            Integer maxDuration = Collections.max(taskDurationList);
            Integer sumDuration =0;
            for (Integer duration : taskDurationList) {
                sumDuration+=duration;
            }
            Integer otherTotalDuration=sumDuration-maxDuration; //其他总耗时
            Integer otherAverageDuration= otherTotalDuration/ (taskDurationList.size()-1);//其他平均耗时
            //最大任务耗时-其他平均任务耗时/ 其他平均耗时  = real percent
            if(assessParam.getTableMetaInfo().getTableName().equals("dws_trade_user_order_td")){
                System.out.println("otherAverageDuration = " + otherAverageDuration);
            }
            BigDecimal realPercent = BigDecimal.valueOf(maxDuration - otherAverageDuration).divide(BigDecimal.valueOf(otherAverageDuration), 2, RoundingMode.HALF_UP).movePointRight(2);


            if(paramPercent.compareTo(realPercent)<0){
                JSONObject skewStageJsonObject = new JSONObject();
                skewStageJsonObject.put("stageId",stageId);
                skewStageJsonObject.put("stageMsg","最大耗时task:"+maxDuration/1000+" 其他平均耗时："+otherAverageDuration/1000);

                skewStagesList.add(skewStageJsonObject);
            }

        }
        if(skewStagesList.size()>0){
            governanceAssessDetail.setAssessScore(BigDecimal.ZERO);
            governanceAssessDetail.setAssessProblem(JSON.toJSONString(skewStagesList));
        }

    }
}
